package github.sf.fw.hbase;

import github.sf.fw.utils.ByteUtil;
import github.sf.fw.utils.StrUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class HbaseKit {
    private final static Logger logger = LoggerFactory.getLogger(HbaseKit.class);
    private final static String CONNECTOR = ":";
    public final static String HBASE_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
    public final static String HBASE_ZOOKEEPER_QUORUM_VAL = "HBASE_ZOOKEEPER_QUORUM";
    public final static String HBASE_ZOOKEEPER_CLIENT_PORT = "hbase.zookeeper.property.clientPort";
    public final static String HBASE_ZOOKEEPER_CLIENT_PORT_VAL = "HBASE_ZOOKEEPER_CLIENT_PORT";
    public final static String HADOOP_HOME = "hadoop.home.dir";
    private final Configuration config;
    private static Connection conn;
    private String namespace = "default";

    private Table currentTable;

    public HbaseKit(HbaseProperties properties) {
        config = HBaseConfiguration.create();
        config.set(HBASE_ZOOKEEPER_QUORUM, System.getProperty(HBASE_ZOOKEEPER_QUORUM));
        config.set(HBASE_ZOOKEEPER_CLIENT_PORT, System.getProperty(HBASE_ZOOKEEPER_CLIENT_PORT));
    }

    public Connection getConn() {
        if (conn == null) {
            try {
                conn = ConnectionFactory.createConnection(config);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return conn;
    }

    public NamespaceDescriptor[] getNamespaces() {
        return doThrow(getHbaseAdmin()::listNamespaceDescriptors);
    }

    public void createNamespace(String name) {
        NamespaceDescriptor build = NamespaceDescriptor.create(name).build();
        doThrowVoid(() -> getHbaseAdmin().createNamespace(build));
    }

    public void deleteNamespace(String name) {
        doThrowVoid(() -> getHbaseAdmin().deleteNamespace(name));
    }

    public List<TableName> getTableNames() {
        HBaseAdmin admin = getHbaseAdmin();
        TableName[] tableNames = doThrow(admin::listTableNames);
        return Arrays.asList(tableNames);
    }

    public HBaseAdmin getHbaseAdmin() {
        HBaseAdmin hBaseAdmin = (HBaseAdmin) doThrow(getConn()::getAdmin);
        try {
            NamespaceDescriptor nsd = hBaseAdmin.getNamespaceDescriptor(namespace);
        } catch (NamespaceNotFoundException e) {
            createNamespace(namespace);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return hBaseAdmin;
    }

    public Table getHTable(TableName tableName) {
        return doThrow(() -> getConn().getTable(tableName));
    }


    public void createTable(String tableName, String[] columnFamilies) {
        HBaseAdmin hbaseAdmin = getHbaseAdmin();
        TableName tableNameObj = TableName.valueOf(tableName);
        if (doThrow(() -> hbaseAdmin.tableExists(tableNameObj))) {
            logger.info("表{}已经存在", tableName);
            return;
        }
        TableDescriptorBuilder.ModifyableTableDescriptor tableDesc = (TableDescriptorBuilder.ModifyableTableDescriptor) TableDescriptorBuilder.newBuilder(tableNameObj).build();
        for (String columnFamily : columnFamilies) {
            ColumnFamilyDescriptor columnFamilyDesc = ColumnFamilyDescriptorBuilder.of(columnFamily);
            tableDesc.setColumnFamily(columnFamilyDesc);
        }
        doThrow(() -> {
            hbaseAdmin.createTable(tableDesc);
            return null;
        });
        close(hbaseAdmin);
    }

    public void deleteTable(String tableName) {
        HBaseAdmin hbaseAdmin = getHbaseAdmin();
        TableName tableNameObj = TableName.valueOf(tableName);
        if (doThrow(() -> hbaseAdmin.tableExists(tableNameObj))) {
            doThrow(() -> {
                hbaseAdmin.disableTable(tableNameObj);
                hbaseAdmin.deleteTable(tableNameObj);
                logger.info("表{}已经删除", tableName);
                return null;
            });
        }
        logger.info("表{}不存在", tableName);
        close(hbaseAdmin);
    }

    public void deleteTable(final HBaseAdmin hbaseAdmin, String tableName) {
        TableName tableNameObj = TableName.valueOf(tableName);
        if (doThrow(() -> hbaseAdmin.tableExists(tableNameObj))) {
            doThrow(() -> {
                hbaseAdmin.disableTable(tableNameObj);
                hbaseAdmin.deleteTable(tableNameObj);
                logger.info("表{}已经删除", tableName);
                return null;
            });
        }
        logger.info("表{}不存在", tableName);
    }

    /**
     * 注意 根据官方文档描述，rowKey尽量不要有序，且columnFamily的数量不要太多，一般两到三个即可
     * 且，family:qualifier的名称不瑶太长，最好不要比value长
     * cell {row,column,version}
     *
     * @param tableName 表名
     * @param rowKey    行key
     * @param family    列族
     * @param qualifier 列族中的具体列
     * @param value     列的具体值
     */
    public void put(String tableName, String rowKey, String family, String qualifier, String value) {
        Table hTable = getHTable(TableName.valueOf(tableName));
        Put put = new Put(ByteUtil.toBytes(rowKey));
        value = checkValue(value);
        put.addColumn(ByteUtil.toBytes(family), ByteUtil.toBytes(qualifier), ByteUtil.toBytes(value));
        doThrowVoid(() -> hTable.put(put));
    }


    public void putAddColumn(Put put, String family, String qualifier, String value) {
        Objects.requireNonNull(put);
        put.addColumn(ByteUtil.toBytes(family), ByteUtil.toBytes(qualifier), ByteUtil.toBytes(value));
    }

    /**
     * 会缓存table，且table不存在时会创建
     *
     * @param tableName 表名
     * @param put       put对象
     * @param family    列簇
     */
    public void put(String tableName, Put put, String family) {
        this.currentTable = getHTable(TableName.valueOf(tableName));
        doThrowVoid(() -> this.currentTable.put(put));
    }

    /**
     * 插入一行的多列数据
     *
     * @param tableName      表面
     * @param rowKey         行key
     * @param family         列簇
     * @param qualifierValue map(qualifier,value)
     */
    public void puts(String tableName, String rowKey, String family, Map<String, String> qualifierValue) {
        Table hTable = getHTable(TableName.valueOf(tableName));
        Put put = new Put(ByteUtil.toBytes(rowKey));
//        Cell cell=new KeyValue();
        for (Map.Entry<String, String> entry : qualifierValue.entrySet()) {
            String value = entry.getValue();
            value = checkValue(entry.getValue());
            put.addColumn(ByteUtil.toBytes(family), ByteUtil.toBytes(entry.getKey()), ByteUtil.toBytes(value));
        }
        doThrowVoid(() -> hTable.put(put));
        close(hTable);
    }

    public void deleteRows(String tableName, String[] rowKeys) {
        Table hTable = doThrow(() -> getHTable(TableName.valueOf(tableName)));
        for (String rowKey : rowKeys) {
            Delete delete = new Delete(ByteUtil.toBytes(rowKey));
            doThrowVoid(() -> hTable.delete(delete));
        }
        close(hTable);
    }

    public Map<String, List<String>> queryTable(String tableName, Scan scan) {
//        Map<String, List<String>> qualifierAndValue = new LinkedHashMap<>();
        Table hTable = getHTable(TableName.valueOf(tableName));
        if (scan == null) {
            scan = new Scan();
        }
        try (ResultScanner scanner = hTable.getScanner(scan)) {
            return extractResult(scanner);
        } catch (IOException e) {
            logger.error("扫描表异常", e);
            throw new RuntimeException(e);
        } finally {
            close(hTable);
        }
    }


    public Map<String, List<String>> queryTableWithPage(String tableName, int page, int size) {
        Table hTable = getHTable(TableName.valueOf(tableName));
        ResultScanner scanner = null;
        try {
            byte[] startRow;
            byte[] stopRow;
            int from = (page - 1) * size;
            startRow = getStartRow(hTable, from);
            int to = page * size;
            stopRow = getStopRow(hTable, to);
            Scan scan = new Scan();
            if (startRow != null && from != 0) {
                scan.withStartRow(startRow, false);
            }
            if (stopRow != null) {
                scan.withStopRow(stopRow, true);
            }
            scanner = hTable.getScanner(scan);
            return extractResult(scanner);
        } catch (IOException e) {
            logger.error("扫描表异常", e);
            throw new RuntimeException(e);
        } finally {
            if (scanner != null) {
                scanner.close();
            }
            close(hTable);
        }
    }

    private Map<String, List<String>> extractResult(ResultScanner scanner) {
        Map<String, List<String>> qualifierAndValue = new LinkedHashMap<>();
        for (Result result : scanner) {
            //遍历列
            for (Cell cell : result.rawCells()) {
                byte[] familyArray = cell.getFamilyArray();
                familyArray = ByteUtil.slice(familyArray, cell.getFamilyOffset(), cell.getFamilyLength());
                byte[] qualifierArray = cell.getQualifierArray();
                qualifierArray = ByteUtil.slice(qualifierArray, cell.getQualifierOffset(), cell.getQualifierLength());
                byte[] valueArray = cell.getValueArray();
                valueArray = ByteUtil.slice(valueArray, cell.getValueOffset(), cell.getValueLength());

                String family = ByteUtil.toString(familyArray);
                String qualifier = ByteUtil.toString(qualifierArray);
                String value = ByteUtil.toString(valueArray);
                String resultKey = resultKey(family, qualifier);
                qualifierAndValue.putIfAbsent(resultKey, new ArrayList<>());
                qualifierAndValue.get(resultKey).add(value);
            }
        }
        return qualifierAndValue;
    }

    public long countTableRow(String tableName) {
        //rowCountByCoprocessor
        Table table = null;
        try {
            long rowCount = 0;

            TableName name = TableName.valueOf(tableName);
            //connection为类静态变量
            table = getHTable(name);
            Scan scan = new Scan();
            //FirstKeyOnlyFilter只会取得每行数据的第一个kv，提高count速度
            scan.setFilter(new FirstKeyOnlyFilter());
            ResultScanner rs = table.getScanner(scan);
            for (Result result : rs) {
                rowCount += result.size();
            }
//            System.out.println("RowCount: " + rowCount);
            return rowCount;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        } finally {
            close(table);
        }
    }


    private <T> T doThrow(ExceptionMethod<T> methodCall) {
        try {
            return methodCall.invokeMethod();
        } catch (Throwable e) {
            logger.error("run method error", e);
            if (RuntimeException.class.isAssignableFrom(e.getClass())) {
                throw (RuntimeException) e;
            }
            throw new RuntimeException(e);
        }
    }

    private void doThrowVoid(ExceptionVoidMethod methodCall) {
        try {
            methodCall.invokeMethod();
        } catch (Throwable e) {
            logger.error("run method error", e);
            if (RuntimeException.class.isAssignableFrom(e.getClass())) {
                throw (RuntimeException) e;
            }
            throw new RuntimeException(e);
        }
    }

    public static String resultKey(String family, String qualifier) {
        return family + CONNECTOR + qualifier;
    }

    public static String[] spiltResultKey(String resultKey) {
        return resultKey.split(CONNECTOR);
    }

    public String checkValue(String value) {
        if (StrUtil.isEmpty(value)) {
            return "";
        }
        return value;
    }

    public String getNamespace() {
        return namespace;
    }

    public void setNamespace(String namespace) {
        this.namespace = namespace;
    }

    public Table getCurrentTable() {
        return currentTable;
    }

    public void close(AutoCloseable closeable) {
        if (closeable != null) {
            try {
                closeable.close();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    private byte[] getStartRow(Table table, int from) {
        byte[] startRow = null;
        Scan scan = new Scan();
        scan.setLimit(from);
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
        Iterator<Result> iterator = scanner.iterator();
        Result result = null;
        while (iterator.hasNext()) {
            result = iterator.next();
        }
        if (result != null) {
            startRow = result.getRow();
        }
        return startRow;
    }

    private byte[] getStopRow(Table table, int to) {
        byte[] stopRow = null;
        Scan scan = new Scan();
        scan.setLimit(to);
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
        Iterator<Result> iterator = scanner.iterator();
        Result result = null;
        while (iterator.hasNext()) {
            result = iterator.next();
        }
        if (result != null) {
            stopRow = result.getRow();
        }
        return stopRow;
    }
}
